package com.millstein.realtime.app.dws;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.PropertyNamingStrategy;
import com.alibaba.fastjson.serializer.SerializeConfig;
import com.millstein.realtime.app.base.BaseAppV1;
import com.millstein.realtime.bean.TradeOrderBean;
import com.millstein.realtime.bean.TradeSkuOrderBean;
import com.millstein.realtime.common.Constants;
import com.millstein.realtime.util.*;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import redis.clients.jedis.Jedis;

import java.math.BigDecimal;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.time.Duration;

public class Dws_09_TradeSkuOrderWindow extends BaseAppV1 {

    public static void main(String[] args) {
        new Dws_09_TradeSkuOrderWindow().init(
                8009,
                3,
                "Dws_09_TradeSkuOrderWindow",
                Constants.TOPIC_DWD_TRADE_ORDER_DETAIL
        );
    }

    /**
     * 具体数据处理的逻辑，由子类编写
     *
     * @param env          执行环境对象
     * @param streamSource 待处理的数据流
     */
    @Override
    protected void handle(StreamExecutionEnvironment env, DataStreamSource<String> streamSource) {
        // 1.去重
        SingleOutputStreamOperator<JSONObject> dataStream = removeDuplicationData(streamSource);
        // 2.构建对象
        SingleOutputStreamOperator<TradeSkuOrderBean> orderStream = buildOrderStream(dataStream);
        // 3.窗口聚合
        SingleOutputStreamOperator<TradeSkuOrderBean> aggregateStream = windowAndAgg(orderStream);
        // 4.补充维度信息
        SingleOutputStreamOperator<TradeSkuOrderBean> resultStream = supplementDim(aggregateStream);
        // 5.写入doris中
        writeToDoris(resultStream);
    }

    /**
     * 使用定时器来对数据进行去重
     * @param streamSource 待去重的数据流
     * @return
     */
    private SingleOutputStreamOperator<JSONObject> removeDuplicationData(
            DataStreamSource<String> streamSource
    ) {
        return streamSource
                .map(new MapFunction<String, JSONObject>() {
                    @Override
                    public JSONObject map(String value) throws Exception {
                        return JSONObject.parseObject(value);
                    }
                })
                .keyBy(item -> item.getString("id"))
                .process(new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                    ValueState<JSONObject> sameDataState;
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        sameDataState = getRuntimeContext().getState(
                                new ValueStateDescriptor<>("same-state", JSONObject.class)
                        );
                    }

                    @Override
                    public void processElement(
                            JSONObject value,
                            Context ctx,
                            Collector<JSONObject> out
                    ) throws Exception {
                        if (sameDataState.value() == null) {
                            // 第一条数据的情况，将当前数据存入状态，然后注册定时器
                            sameDataState.update(value);
                            ctx.timerService().registerProcessingTimeTimer(
                                    ctx.timerService().currentProcessingTime() + 5000
                            );
                        } else {
                            // 非第一条数据的情况，比较状态中数据和当前数据中的row_opt_ts字段，哪个在后面就用哪个
                            String current = value.getString("row_opt_ts");
                            String last = sameDataState.value().getString("row_opt_ts");
                            // 如果current时间在后面，那么更新状态
                            if (CommonUtil.isLarger(current, last)) {
                                sameDataState.update(value);
                            }
                        }
                    }

                    @Override
                    public void onTimer(
                            long timestamp,
                            OnTimerContext ctx,
                            Collector<JSONObject> out
                    ) throws Exception {
                        super.onTimer(timestamp, ctx, out);
                        // 触发定时器，将状态中的数据输出
                        out.collect(sameDataState.value());
                    }
                });
    }

    /**
     * 将数据流转换为TradeSkuOrderBean数据流
     * @param dataStream 待转换的数据流
     * @return
     */
    private SingleOutputStreamOperator<TradeSkuOrderBean> buildOrderStream(
            SingleOutputStreamOperator<JSONObject> dataStream
    ) {
        return dataStream
                .map(new MapFunction<JSONObject, TradeSkuOrderBean>() {
                    @Override
                    public TradeSkuOrderBean map(JSONObject value) throws Exception {
                        return TradeSkuOrderBean.builder()
                                .skuId(value.getString("sku_id"))
                                .skuName(value.getString("sku_name"))
                                .orderOriginTotalAmount(value.getBigDecimal("split_origin_amount"))
                                .orderActivityReduceAmount(
                                        value.getBigDecimal("split_activity_amount") == null
                                                ? BigDecimal.ZERO
                                                : value.getBigDecimal("split_activity_amount")
                                )
                                .orderCouponReduceAmount(
                                        value.getBigDecimal("split_coupon_amount") == null
                                                ? BigDecimal.ZERO
                                                : value.getBigDecimal("split_coupon_amount")
                                )
                                .orderAmount(value.getBigDecimal("split_total_amount"))
                                .ts(value.getLong("od_ts") * 1000)
                                .build();
                    }
                });
    }

    /**
     * 窗口聚合
     * @param orderStream 待聚合的数据流
     * @return
     */
    private SingleOutputStreamOperator<TradeSkuOrderBean> windowAndAgg(
            SingleOutputStreamOperator<TradeSkuOrderBean> orderStream
    ) {
        return orderStream
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<TradeSkuOrderBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner(((element, recordTimestamp) -> element.getTs()))
                )
                .keyBy(TradeSkuOrderBean::getSkuId)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .reduce(
                        new ReduceFunction<TradeSkuOrderBean>() {
                            @Override
                            public TradeSkuOrderBean reduce(
                                    TradeSkuOrderBean bean1,
                                    TradeSkuOrderBean bean2
                            ) throws Exception {
                                bean1.setOrderAmount(bean1.getOrderAmount().add(bean2.getOrderAmount()));
                                bean1.setOrderActivityReduceAmount(
                                        bean1.getOrderActivityReduceAmount().add(bean2.getOrderActivityReduceAmount())
                                );
                                bean1.setOrderCouponReduceAmount(
                                        bean1.getOrderCouponReduceAmount().add(bean2.getOrderCouponReduceAmount())
                                );
                                bean1.setOrderOriginTotalAmount(
                                        bean1.getOrderOriginTotalAmount().add(bean2.getOrderOriginTotalAmount())
                                );
                                return bean1;
                            }
                        },
                        new ProcessWindowFunction<TradeSkuOrderBean, TradeSkuOrderBean, String, TimeWindow>() {
                            @Override
                            public void process(
                                    String key,
                                    Context context,
                                    Iterable<TradeSkuOrderBean> elements,
                                    Collector<TradeSkuOrderBean> out
                            ) throws Exception {
                                TradeSkuOrderBean bean = elements.iterator().next();
                                String startTime = CommonUtil.formatTime(context.window().getStart());
                                String endTime = CommonUtil.formatTime(context.window().getEnd());
                                String curDate = startTime.substring(0, 10);
                                bean.setStt(startTime);
                                bean.setEdt(endTime);
                                bean.setCurDate(curDate);
                                out.collect(bean);
                            }
                        }
                );
    }

    /**
     * 补充维度信息
     * @param aggregateStream 聚合后的数据
     * @return
     */
    private SingleOutputStreamOperator<TradeSkuOrderBean> supplementDim(
            SingleOutputStreamOperator<TradeSkuOrderBean> aggregateStream
    ) {
        return aggregateStream
                .map(new RichMapFunction<TradeSkuOrderBean, TradeSkuOrderBean>() {
                    DruidDataSource dataSource;
                    Jedis redisClient;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        dataSource = DruidDSUtil.getPhoenixConnection();
                        redisClient = RedisUtil.getRedisClient();
                    }

                    @Override
                    public void close() throws Exception {
                        dataSource.close();
                        redisClient.close();
                    }

                    @Override
                    public TradeSkuOrderBean map(TradeSkuOrderBean bean) throws Exception {
                        try {
                            Connection connection = dataSource.getConnection();
                            // 1.SKU
                            JSONObject skuDim = DimUtil.readDim(
                                    connection, redisClient, "dim_sku_info", bean.getSkuId()
                            );
                            bean.setSkuName(skuDim.getString("SKU_NAME"));
                            bean.setTrademarkId(skuDim.getString("TM_ID"));
                            bean.setSpuId(skuDim.getString("SPU_ID"));
                            bean.setCategory3Id(skuDim.getString("CATEGORY3_ID"));

                            // 2.TM
                            JSONObject tmDim = DimUtil.readDim(
                                    connection, redisClient, "dim_base_trademark", bean.getTrademarkId()
                            );
                            bean.setTrademarkName(tmDim.getString("TM_NAME"));

                            // 3.SPU
                            JSONObject spuInfo = DimUtil.readDim(
                                    connection, redisClient, "dim_spu_info", bean.getSpuId()
                            );
                            bean.setSpuName(spuInfo.getString("SPU_NAME"));

                            // 4.CATEGORY3
                            JSONObject c3 = DimUtil.readDim(
                                    connection, redisClient, "dim_base_category3", bean.getCategory3Id()
                            );
                            bean.setCategory3Name(c3.getString("NAME"));
                            bean.setCategory2Id(c3.getString("CATEGORY2_ID"));

                            // 5.CATEGORY2
                            JSONObject c2 = DimUtil.readDim(
                                    connection, redisClient, "dim_base_category2", bean.getCategory2Id()
                            );
                            bean.setCategory2Name(c2.getString("NAME"));
                            bean.setCategory1Id(c2.getString("CATEGORY1_ID"));

                            // 6.CATEGORY1
                            JSONObject c1 = DimUtil.readDim(
                                    connection, redisClient, "dim_base_category1", bean.getCategory1Id()
                            );
                            bean.setCategory1Name(c1.getString("NAME"));

                            connection.close();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }

                        return bean;
                    }
                });
    }

    /**
     * 将结果数据写到doris中
     * @param resultStream 处理完的数据
     */
    private void writeToDoris(SingleOutputStreamOperator<TradeSkuOrderBean> resultStream) {
        resultStream
                .map(item -> {
                    SerializeConfig config = new SerializeConfig();
                    config.propertyNamingStrategy = PropertyNamingStrategy.SnakeCase;
                    return JSONObject.toJSONString(item, config);
                })
                .addSink(FlinkSinkUtil.getDorisSink("gmall.dws_trade_sku_order_window"));
    }
}
